package com.shujia.core.transformations.kv

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object ReduceByKeyOpt {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("groupByKey算子操作")

    val sc = new SparkContext(conf)
    //计算每个人的总分
    val lineRDD: RDD[String] = sc.textFile("spark/data/score.txt")

    //处理lineRDD将其变成KV类型的RDD
    val idWithScoreRDD: RDD[(String, Int)] = lineRDD.map((line: String) => {
      val infos: Array[String] = line.split(",")
      (infos(0), infos(2).toInt)
    })

    /**
     * 做法1：使用GroupByKey分组，后续自定义逻辑聚合
     */
    //使用GroupByKey进行分组
    val idGroupRDD: RDD[(String, Iterable[Int])] = idWithScoreRDD.groupByKey()
    //自定义逻辑聚合
    val resRDD: RDD[(String, Int)] = idGroupRDD.map((kv: (String, Iterable[Int])) => (kv._1, kv._2.toList.sum))
    resRDD.foreach(println)


    /**
     * 做法2：使用reduceByKey直接实现分组聚合
     *
     * reduceByKey只能作用于KV类型的RDD
     * reduceByKey自动的将相同的键分为一组，将来用户只需要传入同一个键对应的值的迭代器中，前一个元素与后一个元素的聚合逻辑即可
     *
     * reduceByKey其中的逻辑包含了预聚合的逻辑
     */
    val resRDD2: RDD[(String, Int)] = idWithScoreRDD.reduceByKey((v1: Int, v2: Int) => v1 + v2)
    resRDD2.foreach(println)

    /**
     *  groupByKey与reduceByKey的区别
     *
     *  1、他们都是只能由KV类型的RDD进行调用
     *  2、groupByKey只注重于分组，至于后续的聚合的逻辑做不了；而reduceByKey分组的同时进行聚合
     *  3、groupByKey中间产生的shuffle数据量要远大于reduceByKey中间产生的shuffle数据量，因为reduceByKey包含了预聚合的逻辑
     *  4、正是因为reduceByKey包含了预聚合的逻辑，所以reduceByKey无法进行一些复杂的逻辑操作，只能做一些简单的聚合操作
     */


    while (true){

    }
  }
}
